package com.joe.xl.devcommon1.util.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
import java.util.Objects;

/**
 * @Author joexl <joe_xuli@163.com>
 * @Since 1.0
 * @Date 2020/10/25
 */
public class JoeProducerInterceptor implements ProducerInterceptor<Object, Object> {
    private volatile long successcnt = 0;
    private volatile long failureCnt = 0;

    @Override
    public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> producerRecord) {
        String prefix = "joe-test";

        return new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(),
                producerRecord.timestamp(), prefix + producerRecord.key(), producerRecord.value(),
                producerRecord.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if(Objects.isNull(e)){
            successcnt++;
        } else {
            failureCnt++;
        }
    }

    @Override
    public void close() {
        System.out.println("---send success- " + successcnt + ", failure- " + failureCnt);
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
